package com.ruoyi.job.service;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Stopwatch;
import com.ruoyi.common.config.DialTestConfig;
import com.ruoyi.common.exception.RequestException;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.reactive.function.BodyExtractors;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;

/**
 * @author Fan Xiao Gang.
 * @ClassName DialTestHandler
 * @Description TOD0
 * @Date 2023/6/29
 */
@Component
@Slf4j
public class DialTestHandler {
    private WebClient webClient;
    public DialTestHandler() {
        webClient = WebClient.builder()
            .clientConnector(new ReactorClientHttpConnector(
                // 允许重定向
                HttpClient.create().followRedirect(true)
            )).build();
    }
    /**
     * @param param
     * @return com.xxl.job.core.biz.model.ReturnT<java.lang.String>
     * @author
     * @date
     */
    @XxlJob("httpDialTestHandler")
    public ReturnT<String> httpDialTestHandler(String param) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        log.info("begin httpDialTestHandler, param: {}", param);
        try {

            DialTestConfig dialTestConfig = JSON.parseObject(param, DialTestConfig.class);
            if (StringUtils.isBlank(dialTestConfig.getUrl())) {
                throw new Exception("url required!");
            }
            if (StringUtils.isBlank(dialTestConfig.getMethod())) {
                dialTestConfig.setMethod("POST");
            }
            Mono<String> resultMono = webClient.method(HttpMethod.valueOf(dialTestConfig.getMethod())).uri(dialTestConfig.getUrl())
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON)
                .bodyValue(Optional.ofNullable(dialTestConfig.getPostBody()).orElse(""))
                .retrieve()
                .onStatus(status -> !status.is2xxSuccessful(), resp -> {
                    Mono<String> body = resp.body(BodyExtractors.toMono(String.class));
                    return body.flatMap(str -> {
                        log.error("request[{}]error，result: {}", dialTestConfig.getUrl(), str);
                        log.error("request[{}]error，result: {}", dialTestConfig.getUrl(), str);
                        // 业务返回的错误
                        if (StringUtils.isNotBlank(str)) {
                            throw new RequestException(str, resp.statusCode().value());
                        }
                        throw new RequestException(resp.statusCode().getReasonPhrase(), resp.statusCode().value());
                    });
                }).bodyToMono(String.class)
                .doOnError(Exception.class, err -> {
                    throw new RequestException(err.getMessage(), -1);
                });
            // 这里不做超时限制，由xxl-job任务的超时时间来打断(interrupt)
            // String result = resultMono.block(Duration.ofSeconds(2000));
            String result = resultMono.block();
            if (!CollectionUtils.isEmpty(dialTestConfig.getResultKeys())) {
                for (String key : dialTestConfig.getResultKeys()) {
                    if (!result.contains(key)) {
                        throw new Exception(String.format("request[%s]error，result not contain: %s", dialTestConfig.getUrl(), key));
                    }
                }
            }
            log.info("httpDialTestHandler success, param: {}", param);
            log.info("httpDialTestHandler success, param: {}", param);
            return ReturnT.SUCCESS;
        } catch (RequestException re) {
            log.error("httpDialTestHandler RequestException, param: {}, code: {}", param, re.getCode(), re);
            log.error("httpDialTestHandler error, param: {}, code: {}, RequestException: {}", param, re.getCode(), re.getMessage());
        } catch (Exception e) {
            log.error("httpDialTestHandler Exception, param: {}", param, e);
            log.error("httpDialTestHandler error, param: {}, Exception: {}", param, e.getMessage());
        } finally {
            stopwatch.stop();
            log.error("finish httpDialTestHandler, param: {}, cost: {}", param, stopwatch.elapsed(
                TimeUnit.MILLISECONDS));
            log.info("finish httpDialTestHandler, param: {}, cost: {}", param, stopwatch.elapsed(TimeUnit.MILLISECONDS));
        }
        return ReturnT.FAIL;
    }
}


